/*
 * Copyright 2023-2025 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.springframework.modulith.events.neo4j;

import static org.assertj.core.api.Assertions.*;
import static org.junit.jupiter.api.Assumptions.*;
import static org.mockito.Mockito.*;

import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.List;
import java.util.UUID;

import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.neo4j.cypherdsl.core.renderer.Dialect;
import org.neo4j.driver.AuthTokens;
import org.neo4j.driver.Driver;
import org.neo4j.driver.GraphDatabase;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.core.env.Environment;
import org.springframework.modulith.events.EventPublication;
import org.springframework.modulith.events.core.EventPublicationRepository;
import org.springframework.modulith.events.core.EventSerializer;
import org.springframework.modulith.events.core.PublicationTargetIdentifier;
import org.springframework.modulith.events.core.TargetEventPublication;
import org.springframework.modulith.events.support.CompletionMode;
import org.springframework.modulith.testapp.TestApplication;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.util.DigestUtils;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.neo4j.Neo4jContainer;
import org.testcontainers.utility.DockerImageName;

/**
 * @author Gerrit Meier
 * @author Cora Iberkleid
 * @author Oliver Drotbohm
 */
@Testcontainers(disabledWithoutDocker = true)
@SpringJUnitConfig(Neo4jEventPublicationRepositoryTest.Config.class)
@ImportAutoConfiguration(classes = Neo4jEventPublicationAutoConfiguration.class)
class Neo4jEventPublicationRepositoryTest {

	static final PublicationTargetIdentifier TARGET_IDENTIFIER = PublicationTargetIdentifier.of("listener");

	@Container //
	static final Neo4jContainer neo4jContainer = new Neo4jContainer(DockerImageName.parse("neo4j:5"))
			.withRandomPassword();

	static abstract class TestBase {

		@Autowired Neo4jEventPublicationRepository repository;
		@Autowired Driver driver;
		@Autowired Environment environment;
		@Autowired EventSerializer eventSerializer;

		CompletionMode completionMode;

		@BeforeEach
		void clearDb() {

			this.completionMode = CompletionMode.from(environment);

			try (var session = driver.session()) {
				session.run("MATCH (n) detach delete n").consume();
			}
		}

		@Test
		void createEventPublication() {

			var testEvent = new TestEvent("id");
			var eventSerialized = "{\"eventId\":\"id\"}";
			var eventHash = DigestUtils.md5DigestAsHex(eventSerialized.getBytes());

			when(eventSerializer.serialize(testEvent)).thenReturn(eventSerialized);
			var publication = repository.create(TargetEventPublication.of(testEvent, TARGET_IDENTIFIER));

			try (var session = driver.session()) {

				var result = session.run("MATCH (p:Neo4jEventPublication) return p")
						.single();

				var neo4jEventPublicationNode = result.get("p").asNode();

				assertThat(UUID.fromString(neo4jEventPublicationNode.get("identifier").asString()))
						.isEqualTo(publication.getIdentifier());
				assertThat(neo4jEventPublicationNode.get("publicationDate").asZonedDateTime().toInstant())
						.isEqualTo(publication.getPublicationDate());
				assertThat(neo4jEventPublicationNode.get("listenerId").asString())
						.isEqualTo(publication.getTargetIdentifier().getValue());
				assertThat(neo4jEventPublicationNode.get("completionDate").isNull()).isTrue();
				assertThat(neo4jEventPublicationNode.get("eventSerialized").asString()).isEqualTo(eventSerialized);
				assertThat(neo4jEventPublicationNode.get("eventHash").asString()).isEqualTo(eventHash);
			}
		}

		@Test
		void updateEventPublication() {

			var testEvent1 = new TestEvent("id1");
			var event1Serialized = "{\"eventId\":\"id1\"}";
			var testEvent2 = new TestEvent("id2");
			var event2Serialized = "{\"eventId\":\"id2\"}";

			when(eventSerializer.serialize(testEvent1)).thenReturn(event1Serialized);
			when(eventSerializer.serialize(testEvent2)).thenReturn(event2Serialized);
			when(eventSerializer.deserialize(event2Serialized, TestEvent.class)).thenReturn(testEvent2);

			var event1 = repository.create(TargetEventPublication.of(testEvent1, TARGET_IDENTIFIER));
			var event2 = repository.create(TargetEventPublication.of(testEvent2, TARGET_IDENTIFIER));

			var now = Instant.now();
			repository.markCompleted(event1, now);

			assertThat(repository.findIncompletePublications()).hasSize(1)
					.element(0)
					.extracting(TargetEventPublication::getEvent).isEqualTo(event2.getEvent());
		}

		@Test
		void findInCompletePastPublications() {

			var testEvent = new TestEvent("id");
			var eventSerialized = "{\"eventId\":\"id\"}";

			when(eventSerializer.serialize(testEvent)).thenReturn(eventSerialized);
			when(eventSerializer.deserialize(eventSerialized, TestEvent.class)).thenReturn(testEvent);

			var event = repository.create(TargetEventPublication.of(testEvent, TARGET_IDENTIFIER));

			var newer = Instant.now().plus(1L, ChronoUnit.MINUTES);
			var older = Instant.now().minus(1L, ChronoUnit.MINUTES);

			assertThat(repository.findIncompletePublicationsPublishedBefore(newer)).hasSize(1)
					.element(0)
					.extracting(TargetEventPublication::getEvent).isEqualTo(event.getEvent());

			assertThat(repository.findIncompletePublicationsPublishedBefore(older)).hasSize(0);
		}

		@Test
		void findIncompleteByEventAndTargetIdentifier() {

			var testEvent = new TestEvent("id");
			var eventSerialized = "{\"eventId\":\"id\"}";

			when(eventSerializer.serialize(testEvent)).thenReturn(eventSerialized);
			when(eventSerializer.deserialize(eventSerialized, TestEvent.class)).thenReturn(testEvent);

			var event = repository.create(TargetEventPublication.of(testEvent, TARGET_IDENTIFIER));

			assertThat(
					repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, event.getTargetIdentifier()))
							.isPresent();
		}

		@Test
		void deletePublicationById() {

			var testEvent = new TestEvent("id");
			var eventSerialized = "{\"eventId\":\"id\"}";

			when(eventSerializer.serialize(testEvent)).thenReturn(eventSerialized);

			var event = repository.create(TargetEventPublication.of(testEvent, TARGET_IDENTIFIER));
			assertThat(repository.findIncompletePublications()).hasSize(1);

			repository.deletePublications(List.of(event.getIdentifier()));
			assertThat(repository.findIncompletePublications()).hasSize(0);
		}

		@Test
		void deleteCompletedPublications() {

			var testEvent1 = new TestEvent("id1");
			var event1Serialized = "{\"eventId\":\"id1\"}";
			var testEvent2 = new TestEvent("id2");
			var event2Serialized = "{\"eventId\":\"id2\"}";

			when(eventSerializer.serialize(testEvent1)).thenReturn(event1Serialized);
			when(eventSerializer.serialize(testEvent2)).thenReturn(event2Serialized);
			when(eventSerializer.deserialize(event1Serialized, TestEvent.class)).thenReturn(testEvent1);
			when(eventSerializer.deserialize(event2Serialized, TestEvent.class)).thenReturn(testEvent2);

			var event1 = repository.create(TargetEventPublication.of(testEvent1, TARGET_IDENTIFIER));

			repository.markCompleted(event1, Instant.now());

			repository.deleteCompletedPublications();

			try (var session = driver.session()) {
				var count = session.run("MATCH (n) WHERE n.completionDate is not null return count(n)").single().get("count(n)")
						.asLong();
				assertThat(count).isEqualTo(0);
			}
		}

		@Test
		void deleteCompletedPublicationsBefore() throws Exception {

			assumeTrue(completionMode == CompletionMode.UPDATE);

			var testEvent1 = new TestEvent("id1");
			var event1Serialized = "{\"eventId\":\"id1\"}";
			var testEvent2 = new TestEvent("id2");
			var event2Serialized = "{\"eventId\":\"id2\"}";

			when(eventSerializer.serialize(testEvent1)).thenReturn(event1Serialized);
			when(eventSerializer.serialize(testEvent2)).thenReturn(event2Serialized);
			when(eventSerializer.deserialize(event1Serialized, TestEvent.class)).thenReturn(testEvent1);
			when(eventSerializer.deserialize(event2Serialized, TestEvent.class)).thenReturn(testEvent2);

			var event1 = repository.create(TargetEventPublication.of(testEvent1, TARGET_IDENTIFIER));

			Instant old = Instant.now();
			repository.markCompleted(event1, old);
			Thread.sleep(100);
			var event2 = repository.create(TargetEventPublication.of(testEvent2, TARGET_IDENTIFIER));
			repository.markCompleted(event2, Instant.now());

			repository.deleteCompletedPublicationsBefore(old.plus(10, ChronoUnit.MILLIS));
			// defensive check just to be sure
			assertThat(repository.findIncompletePublications()).hasSize(0);

			try (var session = driver.session()) {

				var records = session.run("MATCH (n) WHERE n.completionDate is not null return n").list();

				assertThat(records.size()).isEqualTo(1);
				assertThat(records.get(0).get("n").asNode().get("eventSerialized").asString()).contains("id2");
			}
		}

		@Test // GH-451
		void findsCompletedPublications() {

			var event = new TestEvent("first");
			var publication = createPublication(event);

			repository.markCompleted(publication, Instant.now());

			if (completionMode == CompletionMode.DELETE) {

				assertThat(repository.findCompletedPublications()).isEmpty();
				assertThat(repository.findIncompletePublications()).isEmpty();

			} else {

				assertThat(repository.findCompletedPublications())
						.hasSize(1)
						.element(0)
						.extracting(TargetEventPublication::getEvent)
						.isEqualTo(event);
			}
		}

		@Test // GH-258
		void marksPublicationAsCompletedById() {

			var event = new TestEvent("first");
			var publication = createPublication(event);

			repository.markCompleted(publication.getIdentifier(), Instant.now());

			if (completionMode == CompletionMode.DELETE) {

				assertThat(repository.findCompletedPublications()).isEmpty();
				assertThat(repository.findIncompletePublications()).isEmpty();

			} else {

				assertThat(repository.findCompletedPublications())
						.extracting(TargetEventPublication::getIdentifier)
						.containsExactly(publication.getIdentifier());
			}
		}

		@Test // GH-1337
		void countsByStatus() {

			var event = new TestEvent("first");
			var publication = createPublication(event);

			assertOneByStatus(EventPublication.Status.PUBLISHED);

			repository.markFailed(publication.getIdentifier());
			assertOneByStatus(EventPublication.Status.FAILED);

			var resubmitted = repository.markResubmitted(publication.getIdentifier(), Instant.now());

			assertThat(resubmitted).isTrue();
			assertOneByStatus(EventPublication.Status.RESUBMITTED);
		}

		@Test // GH-1337
		void looksUpFailedPublication() {

			var event = new TestEvent("first");
			var publication = createPublication(event);

			repository.markFailed(publication.getIdentifier());

			assertThat(repository.findFailedPublications(EventPublicationRepository.FailedCriteria.ALL))
					.extracting(TargetEventPublication::getIdentifier)
					.containsExactly(publication.getIdentifier());
		}

		@Test // GH-1337
		void claimsResubmissionOnce() {

			var event = new TestEvent("first");
			var publication = createPublication(event);

			repository.markFailed(publication.getIdentifier());

			var now = Instant.now();

			assertThat(repository.markResubmitted(publication.getIdentifier(), now)).isTrue();
			assertThat(repository.markResubmitted(publication.getIdentifier(), now)).isFalse();
		}

		@Test // GH-1337
		void marksPublicationAsProcessing() {

			var event = new TestEvent("first");
			var publication = createPublication(event);

			repository.markProcessing(publication.getIdentifier());
		}

		@Test // GH-1337
		void looksUpFailedPublicationInBatch() {

			var event = new TestEvent("first");
			var publication = createPublication(event);

			repository.markFailed(publication.getIdentifier());

			assertThat(repository.findFailedPublications(EventPublicationRepository.FailedCriteria.ALL.withItemsToRead(10)))
					.extracting(TargetEventPublication::getIdentifier)
					.containsExactly(publication.getIdentifier());
		}

		@Test // GH-1337
		void looksUpFailedPublicationWithReferenceDate() throws Exception {

			var event = new TestEvent("first");
			var publication = createPublication(event);

			repository.markFailed(publication.getIdentifier());

			Thread.sleep(200);

			var criteria = EventPublicationRepository.FailedCriteria.ALL
					.withPublicationsPublishedBefore(publication.getPublicationDate().plusMillis(50));

			assertThat(repository.findFailedPublications(criteria))
					.extracting(TargetEventPublication::getIdentifier)
					.containsExactly(publication.getIdentifier());
		}

		private void assertOneByStatus(EventPublication.Status reference) {

			for (var status : EventPublication.Status.values()) {

				var expected = status == reference ? 1 : 0;

				assertThat(repository.findByStatus(status).size()).isEqualTo(expected);
				assertThat(repository.countByStatus(status)).isEqualTo(expected);
			}
		}

		private TargetEventPublication createPublication(Object event) {

			var token = event.toString();

			doReturn(token).when(eventSerializer).serialize(event);
			doReturn(event).when(eventSerializer).deserialize(token, event.getClass());

			return repository.create(TargetEventPublication.of(event, TARGET_IDENTIFIER));
		}
	}

	@Nested
	@TestPropertySource(properties = CompletionMode.PROPERTY + "=UPDATE")
	class WithUpdateCompletionTest extends TestBase {}

	@Nested
	@TestPropertySource(properties = CompletionMode.PROPERTY + "=DELETE")
	class WithDeleteCompletionTest extends TestBase {}

	@Nested
	@TestPropertySource(properties = CompletionMode.PROPERTY + "=ARCHIVE")
	class WithArchiveCompletionTest extends TestBase {}

	private record TestEvent(String eventId) {}

	@Import({ TestApplication.class })
	@Configuration
	static class Config {

		@Bean
		Driver driver() {
			return GraphDatabase.driver(neo4jContainer.getBoltUrl(),
					AuthTokens.basic("neo4j", neo4jContainer.getAdminPassword()));
		}

		@Bean
		org.neo4j.cypherdsl.core.renderer.Configuration cypherDslConfiguration() {
			return org.neo4j.cypherdsl.core.renderer.Configuration.newConfig().withDialect(Dialect.NEO4J_5).build();
		}

		@Bean
		EventSerializer eventSerializer() {
			return mock(EventSerializer.class);
		}
	}
}
